<!DOCTYPE html>
<head>
    <meta charset="utf-8">

    <script src="/vendor/d3/d3-3.3.4.js"></script>
    <script src="/vendor/dagre-d3/dagre-d3-0.4.14.js"></script>
    <script src="/assets/utils.js"></script>

    <style>
        body {
            font-family: Verdana, sans-serif;
        }
        .clusters rect {
            fill: #e0e0e0;
        }

        .node.stage-stats rect {
            stroke: #999;
            fill: #fff;
            stroke-width: 1px;
        }

        text {
            font-weight: 300;
            font-family: Verdana, sans-serif;
            font-size: 14px;
        }

        .node rect {
            fill: #70c889;
        }

        .edgePath path {
            stroke: #333;
            stroke-width: 2px;
        }
    </style>
</head>

<body>
<svg id="svg-canvas"></svg>
</body>

<script type="text/javascript">

/**
 * TODO: return this from the server directly
 */
function flatten(queryInfo)
{
    var stages = new Map();
    flattenStage(queryInfo.outputStage, stages)

    return {
        id: queryInfo.queryId,
        root: queryInfo.outputStage.plan.id,
        stats: {},
        stages: stages
    }
}

function flattenStage(stageInfo, result)
{
    stageInfo.subStages.forEach(function(stage) {
        flattenStage(stage, result);
    });

    var nodes = new Map();
    flattenNode(result, stageInfo.plan.root, nodes);

    result.set(stageInfo.plan.id, {
        id: stageInfo.plan.id,
        root: stageInfo.plan.root.id,
        distribution: stageInfo.plan.distribution,
        stats: stageInfo.stageStats,
        state: stageInfo.state,
        nodes: nodes
    });
}

function flattenNode(stages, nodeInfo, result)
{
    var sources = [];
    var remoteSources = []; // TODO: put remoteSources in node-specific section
    switch (nodeInfo['@type']) {
        case 'output':
        case 'explainAnalyze':
        case 'project':
        case 'filter':
        case 'aggregation':
        case 'sort':
        case 'markDistinct':
        case 'window':
        case 'rowNumber':
        case 'topnRowNumber':
        case 'limit':
        case 'distinctlimit':
        case 'topn':
        case 'sample':
        case 'tablewriter':
        case 'delete':
        case 'metadatadelete':
        case 'tablecommit':
        case 'groupid':
        case 'unnest':
        case 'scalar':
            sources = [nodeInfo.source];
            break;
        case 'join':
            sources = [nodeInfo.left, nodeInfo.right];
            break;
        case 'semijoin':
            sources = [nodeInfo.source, nodeInfo.filteringSource];
            break;
        case 'indexjoin':
            sources = [nodeInfo.probeSource, nodeInfo.filterSource];
            break;
        case 'union':
        case 'exchange':
            sources = nodeInfo.sources;
            break;
        case 'remoteSource':
            remoteSources = nodeInfo.sourceFragmentIds;
            break;
        case 'tablescan':
        case 'values':
        case 'indexsource':
            break;
        default:
            console.log("unhandled: " + nodeInfo['@type']);
    }

    result.set(nodeInfo.id, {
        id: nodeInfo.id,
        type: nodeInfo['@type'],
        sources: sources.map(function(node) { return node.id }),
        remoteSources: remoteSources,
        stats: {}
    });

    sources.forEach(function(child) {
        flattenNode(stages, child, result);
    });
}

function update(graph, query)
{
    query.stages.forEach(function(stage) {
        var clusterId = "stage-" + stage.id;
        var stageRootNodeId = "stage-" + stage.id + "-root";

        var color = "#f0f0f0";
        switch (stage.state) {
            case "RUNNING":
                color = "#d9edf7";
                break;
            case "FAILED":
                color = "#f2dede";
                break;
            case "ABORTED":
            case "CANCELED":
                color = "#fcf8e3";
                break;
            case "FINISHED":
                color = "#dff0d8";
                break;
        }

        graph.setNode(clusterId, {label: "Stage " + stage.id, clusterLabelPos: 'top-right', style: 'fill: ' + color});

        var stats = stage.stats;
        var html = "" +
                "<div>Output: " + stats.outputDataSize + " / " + formatCount(stats.outputPositions) + " rows</div>" +
                "<hr>" +
                "<div>State: " + stage.state + "</div>" +
                "<div>CPU: " + stats.totalCpuTime + "</div>" +
                (
                        stats.fullyBlocked ?
                        "<div style='color:#f00'>Blocked: " + stats.totalBlockedTime + "</div>" :
                        "<div>Blocked: " + stats.totalBlockedTime + "</div>"
                ) +
                "<div>Memory: " + stats.totalMemoryReservation + "</div>" +
                "<div>Splits: Q:" + stats.queuedDrivers + ", R:" + stats.runningDrivers + ", F:" + stats.completedDrivers + "</div>" +

                "<hr>" +
                "<div>Input: " + stats.processedInputDataSize + " / " + formatCount(stats.processedInputPositions) + " rows</div>";

        graph.setNode(stageRootNodeId, {class: "stage-stats", label: html, labelType: "html"});
        graph.setParent(stageRootNodeId, clusterId);
        graph.setEdge("node-" + stage.root, stageRootNodeId, {style: "visibility:hidden"});

        stage.nodes.forEach(function(node) {
            var nodeId = "node-" + node.id;

            graph.setNode(nodeId, {label: node.type});
            graph.setParent(nodeId, clusterId);

            node.sources.forEach(function(source) {
                graph.setEdge("node-" + source, nodeId, {lineInterpolate: "basis"});
            });

            if (node.type == 'remoteSource') {
                graph.setNode(nodeId, {label: '', shape: "circle"});

                node.remoteSources.forEach(function (sourceId) {
                    graph.setEdge("stage-" + sourceId + "-root", nodeId, {lineInterpolate: "basis"});
                });
            }
        })
    });
}

var graph = new dagreD3.graphlib.Graph({compound: true})
        .setGraph({rankdir: 'BT'})
        .setDefaultEdgeLabel(function () { return {}; });

var render = new dagreD3.render();

var svg = d3.select("#svg-canvas");
var g = svg.append("g");

var first = true;
var refresh = true;

function redraw()
{
    d3.json('/v1/query/' + window.location.search.substring(1), function (error, json) {
        if (error && !refresh) {
            setTimeout(redraw, 1000);
            return;
        }

        update(graph, flatten(json));

        render(d3.select("#svg-canvas g"), graph);

        svg.attr("height", graph.graph().height + 40);
        svg.attr("width", graph.graph().width + 40);
    });
}

setInterval(redraw, 1000);

</script>
